package com.jkl.zookeeperserver.Zookeeper;

import com.jkl.zookeeperserver.Config.ZookeeperConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

@Slf4j
@Component
//服务发现
public class ZookeeperClient {
    final ZookeeperConfig zookeeperConfig;
    private final CuratorFramework curatorFramework;
    private final AtomicReference<MutablePair<String,Integer>> MinMap=new AtomicReference<>(new MutablePair<>(null,Integer.MAX_VALUE));//线程安全,一个键值对
    private final AtomicInteger minConnCount=new AtomicInteger(0);//全局最小连接数量
    private PathChildrenCache cache;

    @Autowired // 确保ZookeeperConfig是通过Spring注入的
    public ZookeeperClient(ZookeeperConfig zookeeperConfig, CuratorFramework curatorFramework) {
        this.zookeeperConfig = zookeeperConfig;
        this.curatorFramework = curatorFramework;
    }
    @PostConstruct
    public void init() {
        try {
            ensurePathExists(zookeeperConfig.getRootNode());// 确保根节点存在
            initNodeWatcher();//初始化zookeeper的节点缓存与添加监听器
        } catch (Exception e) {
            throw new RuntimeException("Failed to initialize ZooKeeper watcher", e);
        }
    }

    private void ensurePathExists(String path) throws Exception {
        if (curatorFramework.checkExists().forPath(path) == null) {
            curatorFramework.create().creatingParentsIfNeeded().forPath(path);
        }
    }
    /**
     *  PathChildrenCache构造函数接收三个参数：
     *  1、CuratorFramework 实例（用于与 ZooKeeper 集群通信）
     *  2、要监听的节点路径（根节点路径）
     *  3、以及一个布尔值（true），指示是否在启动时立即构建初始缓存。
     */
    private void initNodeWatcher() throws Exception {
        // 使用PathChildrenCache来监听子节点及其数据的变化
        cache = new PathChildrenCache(curatorFramework, zookeeperConfig.getRootNode(), true);
        cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);// 加载所有现有的子节点信息，并将其存储在内部缓存中。
        findMinEntry();
        cache.getListenable().addListener((client, event) -> {//开始监听节点的变化
            switch (event.getType()) {
                case CHILD_ADDED://节点的添加
                    minConnCount.set(0);
                    MinMap.set(new MutablePair<>(event.getData().getPath(),0));
                    break;
                case CHILD_UPDATED://节点的数据更新,说明有用户断开连接
                    ChildData data = event.getData();
                    int connCount = Integer.parseInt(new String(data.getData(), StandardCharsets.UTF_8));
                    if(connCount<minConnCount.get()){
                        minConnCount.set(connCount);
                    }
                    if(data.getPath().equals(MinMap.get().getLeft())){
                        MinMap.get().setRight(connCount);//更新连接数
                        if(connCount>minConnCount.get()){
                            findMinEntry();//找下一个最小的
                        }
                    }
                    else{
                        if(connCount<MinMap.get().getRight()){
                            MinMap.set(new MutablePair<>(data.getPath(),connCount));
                        }
                    }
                    break;
                case CHILD_REMOVED://节点移除
                    String path=event.getData().getPath();
                    //重新选举出连接数量最小的节点
                    if(path.equals(MinMap.get().getKey())){
                        if(cache.getCurrentData().isEmpty()){//如果没有其他可选举的节点，那么重置
                            MinMap.set(new MutablePair<>(null,Integer.MAX_VALUE));
                        }
                        else {
                            findMinEntry();
                        }
                    }
                    break;
                default:
                    // 可以忽略其他类型的事件
                    break;
            }
        });
    }

    /**
     * @return 获取连接数最小的节点的数据
     */
    private void findMinEntry(){
        MutablePair<String,Integer> min= MinMap.get();
        for(ChildData childData:cache.getCurrentData()) {
            if(!Objects.equals(MinMap.get().getLeft(),childData.getPath())){
                int connCount = Integer.parseInt(new String(childData.getData(), StandardCharsets.UTF_8));
                if (connCount < min.getRight()) {
                    min.setLeft(childData.getPath());
                    min.setRight(connCount);
                    if(connCount==minConnCount.get()){
                        MinMap.set(min);
                        return ;
                    }
                }
            }
        }
        minConnCount.set(min.getRight());
        MinMap.set(min);
    }

    /**
     * 判断路径是否存在
     * @param path 节点的路径
     * @return true：存在；false：不存在；
     * @throws Exception
     */
    public boolean findPath(String path) throws Exception {
        return curatorFramework.checkExists().forPath(zookeeperConfig.getRootNode() + "/" + path) != null;
    }

    /**
     * 获取要连接数量最少的服务器地址和端口
     * @return 服务器的地址和端口
     */
    public String getPath() {
        return MinMap.get().getLeft().replaceAll("/.*/", "");
    }

    @PreDestroy
    public void destroy() {
        if (curatorFramework != null) {
            curatorFramework.close();
        }
    }
}